package bsf.baseservice.businessmq.systemruntime;


import bsf.base.BsfException;
import bsf.baseservice.businessmq.consumer.ConsumerContext;
import bsf.baseservice.businessmq.dal.tb_consumer_partition_dal;
import bsf.baseservice.businessmq.model.tb_messagequeue_model;
import bsf.baseservice.businessmq.systemruntime.log.ErrorLogHelper;
import bsf.db.DbConn;
import bsf.serialization.json.JsonProvider;
import bsf.util.ConvertUtil;

/**
 MQ消息封装
 
*/
public class MQMessage
{
	/** 
	 转消息为T实体类型
	 
	 <typeparam name="T"></typeparam>
	 @return 
	*/
	public final <T> T getMessageObj(Class<T> cls)
	{

		try
		{
			if (cls == String.class)
			{
				return (T)Model.message;
			}
			else
			{
				return new JsonProvider().<T>deserialize(Model.message,cls);
			}
		}
		catch (Exception exp)
		{
			throw new BsfException("当前消息反序列化失败,json:" + ConvertUtil.objToStr(Model.message), exp);
		}
	}
	/** 
	 基础消息信息
	 
	*/
	public tb_messagequeue_model Model;
	/** 
	 当前消费者上下文
	 
	*/
	public ConsumerContext Context;
	/** 
	 是否已经消费
	 
	*/
	public final boolean getIsMarkFinished()
	{
		return _isMarkFinished;
	}
	private boolean _isMarkFinished = false;
	/** 
	 标记消息为已消费
	 
	*/
	public final void markFinished()
	{
		int consumerclientid = 0;
		int partitionid = 0;
		long mqid = 0;
		try
		{
			MQIDInfo mqidinfo = PartitionRuleHelper.getMQIDInfo(Model.id);
			PartitionIDInfo tempVar = new PartitionIDInfo();
			tempVar.DataNodePartition = mqidinfo.DataNodePartition;
			tempVar.TablePartition =mqidinfo.TablePartition;
			partitionid = PartitionRuleHelper.getPartitionID(tempVar);
			tb_consumer_partition_dal dal = new tb_consumer_partition_dal();
			int result = 0;
			consumerclientid = Context.ConsumerInfo.ConsumerClientModel.id;
			mqid = this.Model.id;
			//尝试标记消息已消费
			boolean isupdatesuccess = false;
			int tryerrorcount = 0;
			while (isupdatesuccess == false)
			{
				try
				{
					try(DbConn c = DbConn.createConn(Context.ConsumerProvider.Config.ManageConnectString))
					{
						result = dal.updateLastMQID(c, consumerclientid, partitionid, mqid);
					}
					isupdatesuccess = true;
				}
				catch (RuntimeException exp)
				{
					tryerrorcount++;
					if (tryerrorcount <= SystemParamConfig.Consumer_TrySetMessageRead_FailCount)
					{
						Thread.sleep((int)(SystemParamConfig.Consumer_TrySetMessageRead_ErrorSleepTime * 1000));
						ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "MarkFinished", String.format("标记消息'已完成'出错,尝试第%1$s次", tryerrorcount + ""), exp);
					}
					else
					{
						throw new BusinessMQException(String.format("标记消息'已完成'出错,放弃第%1$s次重试!", tryerrorcount + ""), exp);
					}
				}
			}
			if (result != 1)
			{
				throw new BusinessMQException(String.format("当前消息消费完成后更新失败,可能是系统级消息乱序消费导致。consumerclientid:%1$s,partitionid:%2$s,mqid:%3$s", Context.ConsumerInfo.ConsumerClientModel.id, partitionid, this.Model.id));
			}
			//消费者端标记为已消费
			_isMarkFinished = true;
		}
		catch (Exception exp)
		{
			throw new BusinessMQException(String.format("当前消息消费标记“已完成”时出错,consumerclientid:%1$s,partitionid:%2$s,mqid:%3$s", consumerclientid, partitionid, mqid), exp);
		}
	}
}